package com.cetc.sdp.kmga.cs.stream.nv;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.cetc.sdp.kmga.cs.jdbc.AuditDAO;
import com.cetc.sdp.kmga.cs.stream.DeviceSetBroadcastWrapper;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;

/**
 * @description:
 * @author： DengQiang
 * @date: 2018/4/3 9:48
 */
public abstract class AbstractDStream {

    public static final String SRC_MESSAGE_KEY = "__source__";

    /**
     * 得到输入流
     * @param jssc JavaStreamingContext
     * @return JavaDStream<String>
     */
    public abstract JavaDStream<String> getDStreamAsPlaintext(JavaStreamingContext jssc);

    public JavaDStream<JSONArray> getPreprocessedJsonDStream(JavaStreamingContext jssc, Set<String> logTypes, boolean saveError, DeviceSetBroadcastWrapper devWrapper, boolean filterDev) {
        Broadcast<Set<String>> logTypesBc = jssc.sparkContext().broadcast(logTypes);
        return getDStreamAsPlaintext(jssc).transform(trans -> {
            Broadcast<Set<String>> devsBc = filterDev ? devWrapper.updateAndGet() : null;
            return trans.map(record -> {
                Map<String, Object> res = new HashMap<>(48);
                try {
                    @SuppressWarnings("unchecked")
                    Map<String, Object> map = (Map<String, Object>) JSON.parse(record.toString());
                    for (Map.Entry<String, Object> entry : map.entrySet()) {
                        res.put(entry.getKey().toLowerCase(), entry.getValue());
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
                res.put(SRC_MESSAGE_KEY, record);
                return res;
            }).filter(m -> {
                Object dev = m.get("devid");
                String logType = String.valueOf(m.get("logtype"));
                boolean flag = logTypesBc.getValue().contains(logType)
                        && dev != null && (!filterDev || devsBc.getValue().contains(dev));
                if (flag) {
                    Object obj = m.get("datas");
                    if (obj instanceof JSONArray) {
                        JSONArray arr = (JSONArray) obj;
                        flag = arr.size() > 0 && arr.get(0) instanceof JSONArray;
                    } else {
                        flag = false;
                        if (saveError) {
                            AuditDAO.saveErrorLog(logType, (String) m.get(SRC_MESSAGE_KEY));
                        }
                    }
                } else {
                    if (saveError) {
                        AuditDAO.saveErrorLog(logType, (String) m.get(SRC_MESSAGE_KEY));
                    }
                }
                return flag;
            }).flatMap(m -> {
                String devId = (String) m.get("devid");
                JSONArray arr = (JSONArray) m.get("datas");
                arr.forEach(x -> ((JSONArray) x).add(devId));
                return arr.iterator();
            }).map(m -> (JSONArray) m).cache();
        });
    }

    /**
     * 每个批次执行完成后调用
     */
    public void afterBatchProcessed() {}
}
